

#include "RequestHandler.h"


RequestHandler::RequestHandler(ConnectionProxy& connection, int connectionId, ConnectionTable* connectionTable, StorexEngine& storexEngine)
: connection(connection),
  resultListener(*this),
  firstSubscriptionDone(false),
  connectionId(connectionId),
  connectionTable(connectionTable),
  storexEngine(storexEngine)
{}

RequestHandler::~RequestHandler(){
/*	if (firstSubscriptionDone) {
		tRexEngine.removeResultListener(&resultListener);
		LOG(info) << "Removed subscriptions for " << connection.remoteToString() << endl;
	}
*/
}

void RequestHandler::handleRequest(std::vector<PktPtr> & pkts){
	for (std::vector<PktPtr>::iterator it= pkts.begin(); it != pkts.end(); it++){
		boost::apply_visitor(PktHandleVisitor(*this), *it);
	}
}

void RequestHandler::PktHandleVisitor::operator()(RulePkt * pkt) const{
/*	LOG(info) << "Rule from " << parent.connection.remoteToString() << ":" << endl
		<< "  " << toString(pkt);
	// Let TRex process (and *delete*) the packet
	parent.tRexEngine.processRulePkt(pkt);
	parent.tRexEngine.finalize();
*/
	parent.storexEngine.processRulePkt(pkt);
}

void RequestHandler::PktHandleVisitor::operator()(PubPkt * pkt) const{
/*	LOG(info) << "Publication from " << parent.connection.remoteToString() << ":" << endl
		<< "  " << toString(pkt);
	// Let TRex process (and *delete*) the packet
	parent.tRexEngine.processPubPkt(pkt);
*/
	parent.storexEngine.processPubPkt(pkt);
}

void RequestHandler::PktHandleVisitor::operator()(SubPkt * pkt) const{

}

void RequestHandler::PktHandleVisitor::operator()(BackSubPkt * pkt) const{
/*	LOG(info) << "Subscription from " << parent.connection.remoteToString() << ":" << endl
		<< "  " << toString(pkt);

	// Check if subscription was already done
	bool alreadySubscribed= false;
	for (vector<SubPkt>::iterator it= parent.subscriptions.begin(); it!= parent.subscriptions.end(); it++){
		if (equals(&(*it), pkt)) {
			alreadySubscribed= true;
			break;
		}
	}
	// If subscription is new add a *copy* of it
	if (!alreadySubscribed) parent.subscriptions.push_back(*pkt);

	delete pkt;

	if (!parent.firstSubscriptionDone){
		parent.tRexEngine.addResultListener(&parent.resultListener);
		parent.firstSubscriptionDone= true;
	}
*/
	parent.connectionTable->registerConnection(parent.connectionId, &parent.resultListener);
	Inbox* inbox = new Inbox(parent.connectionId, pkt);
	parent.storexEngine.processSubPkt(inbox);
}

void RequestHandler::PktHandleVisitor::operator()(UnSubPkt * pUnsub) const{
/*	stringstream log;
	log << "Requested Unsubscription from " << parent.connection.remoteToString() << " for:" << endl
		<< "  " << toString(pUnsub->getSubPkt()) << endl;

	bool wasSubscribed= false;
	for (vector<SubPkt>::iterator subIt= parent.subscriptions.begin(); subIt!= parent.subscriptions.end(); subIt++){
		SubPkt* pSub= &(*subIt);
		if (equals(pSub, pUnsub->getSubPkt())){
			wasSubscribed= true;
			parent.subscriptions.erase(subIt);
			break;
		}
	}

	if (wasSubscribed) log << "  .. subscription removed.";
	else log << "  .. was not subscribed.";
	LOG(info) << log.str();

	delete pUnsub;
*/
	Inbox* inbox = new Inbox(parent.connectionId, pUnsub);
	parent.storexEngine.processUnSubPkt(inbox);
}

void RequestHandler::PktHandleVisitor::operator()(PingPkt * pkt) const{
//	LOG(debug) << "Ping from " << parent.connection.remoteToString();
	delete pkt;
}

void RequestHandler::ResultListenerImpl::handleResult(set<PubPkt *> &genPkts, double procTime){
/*	stringstream log;
	for (set<PubPkt*>::iterator pubIt= genPkts.begin(); pubIt != genPkts.end(); pubIt++){
		PubPkt* pPub= *pubIt;
		for (vector<SubPkt>::iterator subIt= parent.subscriptions.begin(); subIt!= parent.subscriptions.end(); subIt++){
			SubPkt* pSub= &(*subIt);
			if (matches(pSub, pPub)){
				log << endl << "  - " << toString(pPub);
				CharVectorPtr bytes = parent.marshaller.marshal(PktPtr(pPub));
				parent.connection.asyncWrite(bytes);
			}
		}
		// no need to delete the PubPkts: they will be deleted by TRex
	}

	if (log.gcount() > 0){
		LOG(debug) << "Published to " << parent.connection.remoteToString() << ":" << log.str();
	}
*/
	for (set<PubPkt*>::iterator pubIt = genPkts.begin(); pubIt != genPkts.end(); pubIt++){
		PubPkt* pPub= *pubIt;
		PktPtr pkt = PktPtr(pPub);
		CharVectorPtr bytes = parent.marshaller.marshal(pkt);
		parent.connection.asyncWrite(bytes);
	}
}
